package com.bfxy.disruptor.common;

import com.bfxy.disruptor.common.balance.Sequence;
import com.bfxy.disruptor.common.balance.SequenceBarrier;
import com.bfxy.disruptor.common.consume.ConsumeInfo;
import com.bfxy.disruptor.common.consume.EventConsumer;
import com.bfxy.disruptor.common.consume.IConsumerInfo;
import com.bfxy.disruptor.common.consume.EventConsumer;
import com.bfxy.disruptor.common.event.EventHandler;

import java.util.*;
//消费者集合。存储消费者信息的。
public class ConsumerRepository<T> implements Iterable<IConsumerInfo> {
	private final Map<EventHandler<?>, ConsumeInfo<T>> eventProcessorInfoByEventHandler = new IdentityHashMap<EventHandler<?>, ConsumeInfo<T>>();
	private final Map<Sequence, IConsumerInfo> eventProcessorInfoBySequence = new IdentityHashMap<Sequence, IConsumerInfo>();
	private final Collection<IConsumerInfo> consumerInfos = new ArrayList<IConsumerInfo>();

	public void add(final EventConsumer eventprocessor, final EventHandler<? super T> handler,
					final SequenceBarrier barrier) {
		final ConsumeInfo<T> consumerInfo = new ConsumeInfo<T>(eventprocessor, handler, barrier);
		eventProcessorInfoByEventHandler.put(handler, consumerInfo);
		eventProcessorInfoBySequence.put(eventprocessor.getSequence(), consumerInfo);
		consumerInfos.add(consumerInfo);
	}

	public void add(final EventConsumer processor) {
		final ConsumeInfo<T> consumerInfo = new ConsumeInfo<T>(processor, null, null);
		eventProcessorInfoBySequence.put(processor.getSequence(), consumerInfo);
		consumerInfos.add(consumerInfo);
	}

	public void add(final WorkerPool<T> workerPool, final SequenceBarrier sequenceBarrier) {
		final WorkerPoolInfo<T> workerPoolInfo = new WorkerPoolInfo<T>(workerPool, sequenceBarrier);
		consumerInfos.add(workerPoolInfo);
		for (Sequence sequence : workerPool.getWorkerSequences()) {
			eventProcessorInfoBySequence.put(sequence, workerPoolInfo);
		}
	}

	public Sequence[] getLastSequenceInChain(boolean includeStopped) {
		List<Sequence> lastSequence = new ArrayList<Sequence>();
		for (IConsumerInfo consumerInfo : consumerInfos) {
			if ((includeStopped || consumerInfo.isRunning()) && consumerInfo.isEndOfChain()) {
				final Sequence[] sequences = consumerInfo.getSequences();
				Collections.addAll(lastSequence, sequences);
			}
		}

		return lastSequence.toArray(new Sequence[lastSequence.size()]);
	}

	public EventConsumer getEventProcessorFor(final EventHandler<T> handler) {
		final ConsumeInfo<T> eventprocessorInfo = getEventProcessorInfo(handler);
		if (eventprocessorInfo == null) {
			throw new IllegalArgumentException("The event handler " + handler + " is not processing events.");
		}

		return eventprocessorInfo.getEventProcessor();
	}

	public Sequence getSequenceFor(final EventHandler<T> handler) {
		return getEventProcessorFor(handler).getSequence();
	}

	public void unMarkEventProcessorsAsEndOfChain(final Sequence... barrierEventProcessors) {
		for (Sequence barrierEventProcessor : barrierEventProcessors) {
			getEventProcessorInfo(barrierEventProcessor).markAsUsedInBarrier();
		}
	}

	@Override
	public Iterator<IConsumerInfo> iterator() {
		return consumerInfos.iterator();
	}

	public SequenceBarrier getBarrierFor(final EventHandler<T> handler) {
		final IConsumerInfo consumerInfo = getEventProcessorInfo(handler);
		return consumerInfo != null ? consumerInfo.getBarrier() : null;
	}

	private ConsumeInfo<T> getEventProcessorInfo(final EventHandler<T> handler) {
		return eventProcessorInfoByEventHandler.get(handler);
	}

	private IConsumerInfo getEventProcessorInfo(final Sequence barrierEventProcessor) {
		return eventProcessorInfoBySequence.get(barrierEventProcessor);
	}
}
